#include<vector>
#include<string>
#include<functional>
#include<unordered_map>
#include<mutex>
#include<typeinfo>
#include<memory>
#include<thread>
#include<cstdint>
#include<cassert>
#include<cstring>
#include<arpa/inet.h>
#include<netinet/in.h>
#include<unistd.h>
#include<fcntl.h>
#include<sys/epoll.h>
#include<sys/types.h>
#include<sys/socket.h>
#include<sys/eventfd.h>
#include<sys/timerfd.h>

#include"log.hpp"

#define BUFFER_DEFAULT_SIZE 1024
class Buffer
{
private:
    std::vector<char> _buffer; // 使用vector进行空间管理！
    uint64_t _read_idx;        // 读偏移
    uint64_t _write_idx;       // 写偏移
public:
    Buffer()
    :_read_idx(0),_write_idx(0),_buffer(BUFFER_DEFAULT_SIZE) {}

    char* Begin() { return &(*_buffer.begin()); }
    // 1.获取当前写入读取地址
    // buffer的空间起始地址加上写偏移量！
    char* GetWritePosition() { return Begin() + _write_idx; }

    //2.获取当前读取位置起始地址
    char* GetReadPosition() { return Begin() +_read_idx; }

    //3.获取缓冲区末尾空间大小——获取写偏移之后的空闲空间大小   
    uint64_t TailIdleSize() { return _buffer.size() - _write_idx; }

    //4.获取缓冲区起始空间大小——或许读偏移之前的空闲空间大小
    uint64_t HeadIdleSize() { return _read_idx; }

    //5.获取可读数据大小
    uint64_t ReadAbleSize() { return _write_idx - _read_idx; }

    //6.将读偏移向后移动
    void MoveReadOffset(uint64_t len)
    {
        //要小于可读数据大小！否则就读取不到有效数据！
        assert(len>=0);
        // std::cout << "ablesize:" << ReadAbleSize() << " "
        //           << "len: " << len << " index" << " " << _read_idx <<" windex: "<<_write_idx <<std::endl;

        // assert(_read_idx+ len <= ReadAbleSize());这样写不可以！
        // assert(read_idx+ len <= _write_idx);
        assert(len <= ReadAbleSize());
        _read_idx+=len;
    }

    //7.将写偏移向后移动
    void MoveWriteOffset(uint64_t len)
    {
        //向后移动要确保有足够的空间！
        assert(len>=0);
        assert(len <= TailIdleSize());
        _write_idx+=len;
    }

    //8.确保可写空间足够(整体空闲空间足够就移动数据！否则就扩容！)
    void EnsureWriteSpace(uint64_t len)
    {
        assert(len >= 0);
        if (TailIdleSize() >= len)//情况1，后面有足够的空间
        {
            return;
        }
        else if (TailIdleSize() + HeadIdleSize() > len) // 情况2，后面空间不够！但是总体空间足够！
        {
            //将所有的数据进行前挪！
            int pos = _read_idx;
            uint64_t ras = ReadAbleSize();
            std::copy(GetReadPosition(),GetReadPosition()+ras,Begin());//把可读数据拷贝到起始位置！
            //重新设置读写偏移！
            _read_idx = 0;
            _write_idx = ras;
            return;
        }
        else //情况3，空间不够了！直接进行扩容！直接给写偏移之后扩容足够空间即可！
        {
            _buffer.resize(_write_idx+len);
        }

    }

    //9.写入数据
    void Write(void* data,uint64_t len)
    {
        if(len == 0)
            return;
        //1.保证有足够的空间！
        EnsureWriteSpace(len);
        //2.拷贝数据进去
        const char* d = static_cast<const char*>(data);//因为void*进行加法没有意义，所以要进行强转！
        std::copy(d,d+len,GetWritePosition());
    }
    //9.1
    void WriteAndPush(void* data,uint64_t len)
    {
        Write(data,len);
        MoveWriteOffset(len);
    }

    //9.1字符串写入
    void WriteString(const std::string& data)
    {
        Write((void*)data.c_str(),data.size());
    }

    void WriteStringAndPush(const std::string& data)
    {
        WriteString(data);
        MoveWriteOffset(data.size());
    }

    //9.2将另一个buffer的写入到这个buffer里面！
    void WriteBuffer(Buffer& data)
    {
        Write((void*)data.GetReadPosition(),data.ReadAbleSize());
    }

    void WriteBufferAndPush(Buffer& data)
    {
        WriteBuffer(data);
        MoveWriteOffset(data.ReadAbleSize());
    }


    //10.读取数据
    void Read(void* buff,uint64_t len)
    {
        //1.要求获取数据大小，小于可读数据大小！
        assert(len <= ReadAbleSize());
        //2.读取数据
        std::copy(GetReadPosition(),GetReadPosition()+len,(char*)buff);
    }

    //10.1读取数据后将数据弹出出去！
    void ReadAndPop(void* buf,uint64_t len)
    {
        Read(buf,len);
        MoveReadOffset(len);
    }

    //10.2将读取的数据以字符串的方式进行返回
    std::string ReadAsString(uint64_t len)
    {
        assert(len >=0);
        assert(len <= ReadAbleSize());
        std::string str;
        str.resize(len);
        Read((void *)(&str[0]), len);
        //    Read((void*)str.c_str(),len);//c_str返回的是const char*是不允许修改内容的！所以我们不能怎么写！
        return str;
    }

    //10.3
    std::string ReadAsStringAndPop(uint64_t len)
    {
        assert(len <= ReadAbleSize());
        std::string str = ReadAsString(len);
        MoveReadOffset(len);
        return str;
    }

    //11寻找换行回车字符！
    char* FindCRLF()
    {
        void * res = memchr(GetReadPosition(),'\n',ReadAbleSize());//作用是返回这个字符所在的内存地址！
        return (char*)res;
    }
    //12.返回一行数据！这种通常都是针对获取ASCII字符！
    std::string GetLine()
    {
        char* pos = FindCRLF();
        if(pos == NULL)
        {
            return "";
        }
        // std::cout << "读取大小 ："<<pos- GetReadPosition()+1<<std::endl;
        return ReadAsString(pos - GetReadPosition()+1);
        //pos - GetReadPostion的作用就是获取一行的数据！
        //+1是为了将\n也一起获取！
    }

    std::string GetLineAndPop()
    {
        std::string str = GetLine();
        MoveReadOffset(str.size());
        return str;
    }

    //13.清空缓冲区
    void Clear()
    {
        _write_idx = 0;
        _read_idx = 0;
    }
};



class Any
{
private:
    class placeholder
    {
    public:
        virtual ~placeholder() {}
        virtual const std::type_info& type() = 0;
        virtual placeholder* clone() = 0;
    };
    //使用这种方式主要是为了绕过模板的编译期确定类型！
    //让其可以直接定义出来一个any对象，而不是要以any<int>这样的方式才能定义出来一个对象！
    //在需要的时候我们只要通过管理子类holder，通过模板去生成一个我们需要的类型！

    template <class T>
    class holder:public placeholder
    {
    public:
        holder(const T &value)
            : _val(value)
        { }

        virtual const std::type_info& type() //获取子类对象保存的类型！
        {
            return typeid(T);
            //因为type_info的拷贝构造被禁用！只能使用typeid创建出来一个对象！
            // 所以我们必须返回它的引用！
        }

        virtual placeholder* clone()//针对当前的对象自身克隆出一个新的子类对象！
        {
            return new holder(_val);
        }

        T _val;
    };
    placeholder* _content;


public:
    Any():_content(nullptr) {}

    template <class T>
    Any(const T &val)
        : _content(new holder<T>(val))
    {
    }

    Any(const Any &others)//拷贝构造
        : _content(others._content ? others._content->clone() : nullptr)
    {
    }

    ~Any()
    {
        delete _content;
    }

    Any& swap(Any& ohter)
    {
        std::swap(_content,ohter._content);
        return *this;
    }

    template <class T>
    T *get() // 返回保存的数据的指针！
    {
        assert(typeid(T) == _content->type());//获取的数据类型和我们保存的据类型是要一致的！
        return &(((holder<T> *)_content)->_val);

    }

    template<class T>
    Any& operator=(const T& val)//重载赋值运算符，让其可以接收各种类型的赋值！
    {
        //为val构建一个临时通用容器！然后与容器自身进行交换！
        //当出了作用域后，临时对象释放的时候，原先保存的资源也会一起被释放！
        Any(val).swap(*this);
        return *this;
    }

    Any& operator=(Any other)//同类型的赋值！
    {
        //形参本身就是一个临时对象！
        other.swap(*this);
        return *this;
    }
};




#define MAX_LISTEN 1024
class Socket
{
private:
    int _sockfd;
public:
    Socket()
        : _sockfd(-1) {}
    Socket(int fd)
        : _sockfd(fd) {}
    ~Socket(){Close();}

    int FD(){return _sockfd;}

    //创建套接字
    bool Create()
    {
        _sockfd = socket(AF_INET,SOCK_STREAM,IPPROTO_TCP);
        if(_sockfd < 0)
        {
            ERR_LOG("SOCKFD CREAT FAILED!");
            return false;
        }
        DBG_LOG("SOCKFD CREAT Success!");
        return true;
    }

    // 绑定地址信息
    bool Bind(const std::string& ip,uint16_t port)
    {
        struct sockaddr_in addr;
        bzero(&addr,sizeof(addr));

        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(struct sockaddr_in);

        int n = bind(_sockfd,(struct sockaddr*)&addr,len);
        if(n < 0) 
        {
            ERR_LOG("BIND ADDRESS FAILED!");
            return false;
        }
        DBG_LOG("BIND SUCESS!");
        return true;
    }

    // 开始监听
    bool Listen(int backlog = MAX_LISTEN)
    {
       int n = listen(_sockfd,backlog);
       if(n < 0)
       {
           ERR_LOG("LIETEN FAILED!");
           return false;
       }
       DBG_LOG("LISTEN SUCESS!");
       return true; 
    }

    // 向服务器发起连接
    bool Connect(const std::string& ip,uint16_t port)
    {
        struct sockaddr_in addr;
        bzero(&addr,sizeof(addr));

        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(addr);

        if(connect(_sockfd,(struct sockaddr*)&addr,len) < 0) 
        {
            ERR_LOG("CONNECT SERVER FAILED!");
            return false;
        }
        return true;
    }

    // 获取新连接
    int Accept()
    {
        int newfd = accept(_sockfd,nullptr,nullptr);
        if(newfd < 0)
        {
            ERR_LOG("SOCKET ACCEPT FAILED!");
            return -1;
        }
        return newfd;
    }

    // 接收数据
    ssize_t Recv(void *buf,size_t len,int flag = 0)//flag是用来进行设设置是阻塞还是非阻塞！
    {
        ssize_t ret = recv(_sockfd,buf,len,flag);
        if(ret <= 0)
        {
            if(errno ==  EAGAIN || errno == EINTR)
            {
                //EAGAIN 表示缓冲区没有数据！直接返回在非阻塞的情况下才会出现！
                //EINTR 表示当前的阻塞等待被信号打断！
                return 0;//表示这次没有接收到数据！但是连接没有出错！
            }
            ERR_LOG("SOCKET RECV FAILED!");
            return -1;
        }
        return ret;
    }
    ssize_t NonBlockRecv(void* buf,size_t len)
    {
        return Recv(buf,len,MSG_DONTWAIT);//表示当前进接收为非阻塞！
    }

    // 发送数据
    ssize_t Send(void* buf,ssize_t len,int flag =0)
    {
        ssize_t ret = send(_sockfd,buf,len,flag);
        if(ret < 0)
        {
            if(errno ==  EAGAIN || errno == EINTR)
            {
                return 0;
            }
            ERR_LOG("SOCKET SEND FAILED!");
            return -1;
        }
        return ret;//实际发送的数据长度！

    }
    ssize_t NonBlockSend(void* buf,ssize_t len)
    {
        if(len == 0) return 0;
        return Send(buf,len,MSG_DONTWAIT);//发送为非阻塞！
    }

    // 关闭套接字
    void Close()
    {
        if(_sockfd != -1)
            close(_sockfd);
            _sockfd = -1;
    }

    // 创建一个服务端连接
    bool CreateServer(uint16_t port, const std::string &ip = INADDR_ANY, bool block_flag = false)
    {
        // 1.创建套接字
        if (Create() == false)
            return false;
        // 2.设置非阻塞属性！
        if (block_flag)
            NonBlock();

        // 3.启动地址重用
        ReuseAddress();
        // 4.绑定套接字
        if (Bind(ip, port) == false)
            return false;
        // 5.开始监听
        if (Listen() == false)
            return false;
        return true;
    }
    // 创建一个客户端连接
    bool CreateClient(uint16_t port, const std::string &ip)
    {
        // 1.创建套接字
        if (Create() == false)
            return false;
        // 2.连接服务器！
        if (Connect(ip, port) == false)
            return false;
        return true;
    }

    // 设置套接字选项---开启地址端口重用
    void ReuseAddress()
    {
        int opt =1;
        setsockopt(_sockfd,SOL_SOCKET,SO_REUSEADDR,&opt,sizeof(opt));
        opt =1;
        setsockopt(_sockfd,SOL_SOCKET,SO_REUSEPORT,&opt,sizeof(opt));
        DBG_LOG("REUSE SUCESS!");
    }
    // 设置套接字阻塞属性--设置为非阻塞
    void NonBlock()
    {
        int fl = fcntl(_sockfd,F_GETFL);
        if (fl < 0) { ERR_LOG("NONBLOCK ERROR!"); }
        fcntl(_sockfd, F_SETFL, fl | O_NONBLOCK);
    }

};

class Poller;
class EventLoop;
class Channel
{
private:
    int _fd;
    uint32_t _events;  // 当前要监控的事件
    uint32_t _revents; // 当前连接触发的事件
    EventLoop* _loop;
    using EventCallback = std::function<void()>;
    EventCallback _read_callback;  // 可读事件被触发的回调函数！
    EventCallback _write_callback; // 可写事件被触发的回调函数！
    EventCallback _error_callback; // 错误事件被触发的回调函数！
    EventCallback _close_callback; // 连接关闭事件被触发的回调函数!
    EventCallback _event_callback; // 任意时间被触发的回调函数！
public:
    Channel(EventLoop* loop,int fd)
    :_fd(fd),_events(0),_revents(0),_loop(loop)
    {}


    //获取文件描述符
    int FD()
    {
        return _fd;
    }
    uint32_t Events()//获取想要监控的事件！
    {
        return _events;
    }

    void SetRevents(uint32_t events)//设置实际就绪的事件！
    {
        _revents = events;
    }
    // 设置事件回调
    void SetReadCallback(const EventCallback &cb)
    {
        _read_callback = cb;
    }
    void SetWriteCallback(const EventCallback &cb)
    {
        _write_callback = cb;
    }
    void SetErrorCallback(const EventCallback &cb)
    {
        _error_callback = cb;
    }
    void SetCloseCallback(const EventCallback &cb)
    {
        _close_callback = cb;
    }
    void SetEventCallback(const EventCallback &cb)
    {
        _event_callback = cb;
    }



    // 当前该事件是否是监控了可读的！！
    bool ReadAble()
    {
        //是否可读就是查看当前是否监控了可读事件！
        return _events & EPOLLIN;
    }

     // 当前事件是否监控了可写！
    bool WriteAble() 
    {
        return _events & EPOLLOUT;
    }

    void EnableRead() // 启动读事件监控
    {
        _events|=EPOLLIN;
        Update();
    }

    void EnableWrite() // 启动写事件监控
    {
        _events |= EPOLLOUT;
        Update();

    }
    void DisableRead() // 解除读事件监控
    {
        _events &= ~EPOLLIN;//这样子就可以取消我们原先的监控！
        Update();
    }
    void DisabelWrite() // 解除写事件监控
    {
        _events &= ~EPOLLOUT;
        Update();
    }
    void DisableAll() // 关闭所有事件监控
    {
        _events = 0;
        Update();
    }
    void Remove();// 移除监控——从epoll中彻底移除！

    void Update();
    //不能再类里面实现！因为涉及到了poller类里面的内部函数！
    //我们只是声明了类！但是编译器不知道类里面有什么成员！必须放在类外进行实现！


    void HandleEvent() // 事件处理函数！一旦连接触发了事件就调用这个函数！触发了什么事件就自己决定调用什么函数
    {
        //EPOLLRDHUP是指对方关闭了连接！要求上层对缓冲区数据进行处理！因为对方已经不可能发送信的数据了！
        //EPOLLPRI是优先数据！只带外数据！
        //
        if((_revents & EPOLLIN )|| (_revents & EPOLLRDHUP )||(_revents & EPOLLPRI))
        {
            if (_event_callback) // 只要有那么就要调用！
                _event_callback();
            if(_read_callback)
                _read_callback();
        }

        else if (_revents & EPOLLOUT)
        {
            if (_event_callback) // 只要有就要调用！在事件处理完毕后调用！例如：用来刷新活跃度！如果放在之前可能会导致时间处理完毕后活跃度就已经没有了！
                _event_callback();

            if (_write_callback)
                _write_callback();

        }
        else if(_revents & EPOLLERR)
        {
            if (_event_callback) // 只要有，就都要调用！
                _event_callback();

            //因为出现异常就会导致释放！所以任意回调要放在前面调用！
            if (_error_callback)
                _error_callback();
        }
        else if(_revents & EPOLLHUP)//表示channel里面对应的文件描述符已经被关闭了
        {
            if (_event_callback) // 只要有，就都要调用！
                _event_callback();

            if(_close_callback)
                _close_callback();
        }
        //这种有可能让连接断开的事件一次只能处理一个！
        //例如：写入事件出错！那么就要断开连接！那么后面的错误处理和异常处理就不能进行了？！
    }
};


#define MAX_EPOLLEVENTS 1024
class Poller
{
private:
    int _epfd;
    struct epoll_event _evs[MAX_EPOLLEVENTS];
    std::unordered_map<int,Channel*> _channels;
private:
    //对epoll的直接操作！
    void Update(Channel* channel,int op)
    {
        int fd = channel->FD();
        struct epoll_event evt;
        evt.data.fd = fd;
        evt.events = channel->Events();
        int ret = epoll_ctl(_epfd,op,fd,&evt);
        if(ret < 0)
        {
            ERR_LOG("EPOLL CONTROL FAILED!");
            abort();
        }
        return;
    }

    //判断Channel是否已经添加监控！
    bool HasChannel(Channel* channel)
    {
        auto it = _channels.find(channel->FD());
        if(it == _channels.end())
        {
            return false;
        }
        return true;
    }

public:
    Poller()
    {
        _epfd = epoll_create(MAX_EPOLLEVENTS);
        if(_epfd < 0)
        {
            ERR_LOG("EPOLL CREATE FAILED!");
            abort();
        }
    }
    ~Poller()
    {}
    //添加或者修改监控事件
    void UpdateEvent(Channel* channel)//我们是直接对Channel进行操作的！Channel里面有fd也有其对应的事件！
    {
        bool ret = HasChannel(channel);
        if(ret == false)//不存在就添加
        {
            Update(channel,EPOLL_CTL_ADD);
            _channels.insert({channel->FD(),channel});
        }
        else//存在就更新！
        {
            Update(channel, EPOLL_CTL_MOD);
        }
    }


    //移除监控
    void ReomveEvent(Channel* channel)
    {
        bool ret = HasChannel(channel);
        if (ret == true)
        {
            auto it = _channels.find(channel->FD());
            if(it != _channels.end())
            {
                _channels.erase(it);
            }
            Update(channel, EPOLL_CTL_DEL);
        }
    }

    void Poll(std::vector<Channel*>* active)//有哪些Channel现在是活跃的！要将其返回出去
    {
        int nfds = epoll_wait(_epfd,_evs,MAX_EPOLLEVENTS,-1);
        if(nfds < 0)
        {
            if(errno == EINTR)//被信号中断！
            {
                return;
            }
            ERR_LOG("EPOLL_WAIT FAILED! :%s",strerror(errno));
            abort();
        }
        for(int i = 0;i<nfds;i++)
        {
            auto it = _channels.find(_evs[i].data.fd);//找到这个fd对应的Channel
            assert(it != _channels.end());
            it->second->SetRevents(_evs[i].events);//在fd对应的Channel里面设置实际就绪的事件！
            active->push_back(it->second);
        }
    }
    // 对端正常关闭（程序里close()，shell下kill或ctr+c），触发EPOLLIN和EPOLLRDHUP，但是不触发EPOLLERR 和EPOLLHUP。——重点！
};




using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>;

class TimeTask{

private:
    uint64_t _id;      // 能够被找到这个定时器任务！因为id可能会很多所以使用uint64_t
    uint32_t _timeout; // 定时任务的超时时间！——决定了什么是被执行！
    TaskFunc _task_cb; // 定时器对象要执行的任务！——放在析构函数里面的！
    ReleaseFunc _release;//删除TimerWheel中的map中的保存的定时器的信息！防止内存泄漏！
    bool _canceled;//false表示没有被取消，true表示取消！

public:
    TimeTask(uint64_t id, uint32_t delay, const TaskFunc &cb)
        : _id(id), _timeout(delay), _task_cb(cb),_canceled(false)
    {}

    ~TimeTask()
    {
        if (_canceled == false)//任务取消要放在内部！而不是放在时间轮！如果放在时间轮让其销毁！那就是提前执行任务！
        {
            _task_cb(); // 销毁的时候执行任务
        }
        _release();
    }

    void SetRelease(const ReleaseFunc& cb)//这个释放函数由外部进行设置！
    {
        _release = cb;
    }

    void CanCeled(uint64_t id) { _canceled = true; }//取消任务

    uint32_t DelayTime() { return _timeout; }
};


class EventLoop;

class TimeWheel
{
    using PtrTask = std::shared_ptr<TimeTask>;
    using WeakTask= std::weak_ptr<TimeTask>;
private:
    int _capacity;//表盘的最大数量！——就是最大延迟时间！
    std::vector<std::vector<PtrTask>>  _wheel;//二维数组
    int _tick;//走到哪里释放哪里的对象！——是时间轮的秒针！
    std::unordered_map<uint64_t,WeakTask> _timers;
    //当我们要二次添加同一个定时器任务对象的时候，我们首先要找它们共同的一个计数器！
    //但是我们又不能重新用shared_ptr保存一份，放在map里面用于查找！因为这样子会导致计数器+1
    //即使数组里面的share_ptr都释放掉了！但是因为map里面还有一份！
    //所以这就会让其一直无法释放！所以我们必须使用weak_ptr让其不参与计数！
    //然后我们使用这个weak_ptr去构造shared_ptr这样子就可以让后面构建的对象共享同一个计数!

    //我们还要考虑到线程安全问题！因为如果多个线程同时操作TimeWheel向_timers添加移除数据，那么就会有线程安全的问题！
    //所以我们要保证所有的操作都在_loop相关的线程里面统一进行完成！

    EventLoop* _loop;
    int _timerfd;//定时器描述符
    //每到期一次就执行一次过期任务！
    std::unique_ptr<Channel>  _timer_channel;//定时器描述符关联的channel
    //定时器描述符的可读事件回调就是读取计时器，然后执行定时任务！


private:
    void RemoveTimer(uint64_t id)
    {
        auto it = _timers.find(id);
        if(it != _timers.end())
        {
            _timers.erase(id);
        }
    }

    static int CreateTimerFd()
    {
        int timerfd = timerfd_create(CLOCK_MONOTONIC,0);
        if(timerfd < 0)
        {
            ERR_LOG("TIMERFD CREATE FAILED!");
            abort();
        }
        struct itimerspec itime;
        itime.it_value.tv_sec =1;
        itime.it_value.tv_nsec = 0;//设置第一次超时时间为1s

        itime.it_interval.tv_sec = 1;
        itime.it_interval.tv_nsec = 0;//设置第一次超时后，每次超时间隔多少秒！
        timerfd_settime(timerfd,0,&itime,NULL);
        return timerfd;

    }
    void  RunTimerTask()//这个函数是每秒钟执行一次！相当于秒针走一次！
    {
        _tick = (_tick+1) % _capacity;
        _wheel[_tick].clear();//清空指定位置的数组就会将数组保存的所有管理定时器对象的shared_ptr释放掉！
    }

    void ReadTimerfd()
    {
        uint64_t times;
        int ret = read(_timerfd,&times,8);
        if (ret < 0)
        {
            ERR_LOG("READ TIMERFD FAILED");
            abort();
        }
        return;
    }

    void OnTime()
    {
        ReadTimerfd(); // 要将内容拿出来防止一直通知
        RunTimerTask();
    }

    void TimerAddInLoop(uint64_t id, uint32_t delay, const TaskFunc &cb)//添加定时任务！
    {
        PtrTask pt(new TimeTask(id,delay,cb));
        pt->SetRelease(std::bind(&TimeWheel::RemoveTimer,this,id));
        //当定时任务对象被销毁的时候，要从map里面将信息移除！
        //因为RemoveTimer有一个隐藏参数this和Release本身要求无参！所以我们可以使用bind来解决这个问题！

        int pos = (_tick + delay) % _capacity; // 防止越界！
        _wheel[pos].push_back(pt);//找到的位置是一数组！要插入数组中！

        _timers[id] = WeakTask(pt);
        //将本次要添加的TimeTask对象放入
        //要放入的是Weakptr而不是shared_ptr!
    }

    void TimerRefreshInLoop(uint64_t id)//刷新定时任务！
    {
        //刷新的逻辑就是通过id来查找到那个weak_Ptr，然后通过weak_ptr来实例化一个新的shared_ptr然后添加进时间轮中！
        auto it = _timers.find(id);
        if(it == _timers.end())
        {
            return;//没找到,就直接结束！定时任务无法刷新！
        }
        //找到构建一个shared_ptr出来！ 
        PtrTask pt = it ->second.lock();//lock获取weak_ptr管理的对象的对应的shared_ptr
        int delay = pt->DelayTime();
        int pos = (_tick + delay) % _capacity; // 防止越界！
        _wheel[pos].push_back(pt);//找到的位置是一数组！要插入数组中！
        //指向当前秒针指向的位置加上延迟的位置！
    }

    void TimeCancelInLoop(uint64_t id)
    {
        auto it = _timers.find(id);
        if(it == _timers.end())
        {
            return;//没找到,就直接结束！定时任务无法刷新！
        }
        PtrTask pt = it ->second.lock();
        pt->CanCeled(id);
    }

public:

    TimeWheel(EventLoop* loop)
        : _capacity(60), _tick(0),_wheel(_capacity),
        _loop(loop),_timerfd(CreateTimerFd()),
        _timer_channel(new Channel(_loop,_timerfd))

    {
        _timer_channel->SetReadCallback(std::bind(&TimeWheel::OnTime,this));
        _timer_channel->EnableRead();//启动读事件监控！
    }

    //这个接口存在线程安全问题！所以也要在EventLoop里面调用！
    bool HasTimer(uint64_t id)
    {
        auto it = _timers.find(id);
        if (it == _timers.end())
        {
            return false; // 没找到,就直接结束！定时任务无法刷新！
        }
        return true;
    }

    /*定时器中有一个_timers成员，定时器信息的操作可能会存在多个线程中！因此要考虑线程安全的问题！*/
    /*因为我们不想要加锁！这样会导致效率下降所以我们将所有的操作放在一个线程中进行！*/
    void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc &cb);//添加定时任务！

    void TimerRefresh(uint64_t id);//刷新定时任务！

    void TimeCancel(uint64_t id);


    //因为里面使用到了EventLoop类的函数！所以要放在EventLoop下面进行定义！
    //我们只是告诉编译器有这个类！但是编译器也不知道有什么函数！


};


class EventLoop
{
private:
    std::thread::id _thread_id;//这是一个线程id，用于区分是不是在EvenLoop所在的线程池！在就是直接执行！不在就压入任务池！
    int _event_fd;//事件通知描述符！通过它来唤醒IO事件有可能导致的阻塞！
    Poller _poller;//进行所有描述符的事件监控
    std::unique_ptr<Channel> _event_channel;
    using Functor = std::function<void()>;
    std::vector<Functor> _task;//任务池
    std::mutex _mutex;//保证任务池操作的线程安全！
    TimeWheel _time_wheel;
private:
    void RunAllTask()//执行任务池中的所有任务
    {
        std::vector<Functor> functors;
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _task.swap(functors);//在加锁期间将_task里面的所有任务拿出来！
            //之所以怎么做是隐如果我们直接给_task加锁会导致一个问题
            //那就是执行速度慢！那样子如果在执行任务期间，有新的任务想要插入，就会阻塞!
            //所以为了防止这样的问题，我们可以直接将任务都置换到另一个vector里面！
            //那么加锁的保护的只需要这个操作就可以了！
            //而任务就交给这个新的vector来执行！
        }
        for(auto& f : functors)
        {
            f();
        }
        return;
    }
    static int CreateEventFd()
    {
        int efd =  eventfd(0,EFD_CLOEXEC|EFD_NONBLOCK);
        if(efd < 0)
        {
            ERR_LOG("eventfd failed!");
            abort();
        }
        return efd;
    }
    void ReadEventFd()
    {
        uint64_t res = 0;
        int ret = read(_event_fd,&res,sizeof(res));
        if(ret < 0)
        {
            if (ret == EINTR || ret == EAGAIN)
                return;
            else
            {
                ERR_LOG("READ EVENTFD FAILED");
                abort();
            }
        }
        return;
    }
    void WeakUpEventFd()
    {
        uint64_t val = 0;
        int ret = write(_event_fd,&val,sizeof(val));
        if(ret < 0)
        {
            if (ret == EINTR)//被信号打断
                return;
            else
            {
                ERR_LOG("WRITE EVENTFD FAILED");
                abort();
            }
        }
        return;

    }
public:
    EventLoop()
        : _thread_id(std::this_thread::get_id()),
          _event_fd(CreateEventFd()),
          _event_channel(new Channel(this,_event_fd)),
          _time_wheel(this)
    {
        //给eventfd添加可读事件回调函数，读取eventfd事件通知的次数！
        _event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventFd,this));
        //启动读事件监控！
        _event_channel->EnableRead();
    }

    void Start()//启动模块！首先进行事件监控,然后处理事件，事件处理完后开始执行任务！
    {
        std::vector<Channel*> actives;
        _poller.Poll(&actives);
        //poller的文件描述符哪里哪里来的呢？
        //只要绑定的是同一个poller那么，外部channel调用EnableRead胡总EnableWrite的时候都会添加进来！
        for(auto a:actives)//处理事件
        {
            a->HandleEvent();
        }
        RunAllTask();//执行任务
    }

    bool IsInLoop()//判断是不是再EventLoop这个线程里面！
    {
        //this_thread::get_id是获取当前线程运行的id！
        //_thread_id是EventLoop所在线程的id
        return _thread_id == std::this_thread::get_id();
    }

    void AssertInLoop()
    {
        assert(std::this_thread::get_id() == _thread_id);
    }




    void QueueInLoop(const Functor &cb) // 将操作压入任务池中！
    {
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _task.push_back(cb);
            //关于task的操作都要加锁！因为是属于共享资源！
        }
        //为了防止没有事件就绪导致的EventLoop一直监控poller.poll()——这个函数的阻塞！
        // 从而导致任务无法得到执行！我们就必须唤醒任务！

        WeakUpEventFd();
        //唤醒其实就是写入数据，让Eventfd这个事件就绪！

    }

    //给EventLoop提供一个任务让其执行！
    //然后判断，如果是在线程内就直接执行！如果不在就压入队列中！
    void RunInLoop(const Functor& cb)
    {
        if(IsInLoop() == true)
            cb();//直接执行任务！
        else
            QueueInLoop(cb);//将任务压入任务池中！
    }

    void UpdateEvent(Channel* channel)//添加或者修改描述符的监控事件
    {
        _poller.UpdateEvent(channel);
    }

    void RemoveEvent(Channel* channel)//移除描述符的监控事件！
    {
        _poller.ReomveEvent(channel);
    }

    void TimerAdd(uint64_t id,uint32_t delay,const TaskFunc& cb)
    {
        _time_wheel.TimerAdd(id,delay,cb);
    }

    void TimerRefresh(uint64_t id) // 刷新定时任务！
    {
        _time_wheel.TimerRefresh(id);
    }

    void TimeCancel(uint64_t id)
    {
        _time_wheel.TimeCancel(id);
    }

    bool HasTimer(uint64_t id)//是否存在某个定时器！
    {
        return _time_wheel.HasTimer(id);
    }
};



void Channel::Remove() 
{
    _loop->RemoveEvent(this);
}

void Channel::Update()
{
    _loop->UpdateEvent(this);
    
}

void TimeWheel:: TimerAdd(uint64_t id, uint32_t delay, const TaskFunc &cb) // 添加定时任务！
{
    // 因为很多定时任务都设计到连接的操作！所以要考虑到线程安全！
    // 所以添加操作要放在EventLoop线程里面执行！
    _loop->RunInLoop(std::bind(&TimeWheel::TimerAddInLoop, this, id, delay, cb));
}

void TimeWheel::TimerRefresh(uint64_t id) // 刷新定时任务！
{
    _loop->RunInLoop(std::bind(&TimeWheel::TimerRefreshInLoop, this, id));
}

void TimeWheel::TimeCancel(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimeWheel::TimeCancelInLoop, this, id));
}

typedef enum
{
    DISCONNECTED, // 连接关闭，处于这个状态就不能进行操作！然后开始释放资源！
    CONNECTING,   // 连接建立成功-待处理状态
    CONNECTED,    // 连接设置已经完成—可以通信的状态
    DISCONNECTING // 待关闭状态！例如：当我们关闭的时候，设置为这个状态，等到所有的数据处理完毕就设置为DISCONNECTED

} ConnStatus;


class Connection;
using PtrConnection = std::shared_ptr<Connection>;
class Connection:public std::enable_shared_from_this<Connection>
{
private:
    uint64_t _conn_id;             // 连接的唯一id，便于连接的管理的和查找！
    // uint64_t _timer_id;            // 定时器id,必须是唯一的！上面的定时器配合使用！但是为了简化操作我们使用conn_id来进行代替！
    int _sockfd;                   // 连接关联的文件描述符！
    EventLoop *_loop;              // 连接锁关联的loop！
    bool _enable_inactive_release; // 判断连接是否启动非活跃销毁的标志！默认为false
    ConnStatus _status;            // 保存连接的状态！
    Socket _socket;                // 通过这个对象来管理套接字的各种操作！
    Channel _channel;              // 对连接的事件的管理！
    Buffer _in_buffer;             // 输入缓冲区！——从Socket中读取到的数据！
    Buffer _out_buffer;            // 输出缓冲区！——存放要发送给对端的数据！
    Any _context;
    // 因为我们的服务器是支持多种协议的！
    // 如果要切换协议那么我们就要保存上下文！记录处理请求处于的哪一个阶段！以便于有新数据来了，可以接着上一次的处理继续执行！
    // 而不同的协议数据格式是不一样的！所以我们就必须使用通用的数据结构来进保存！

    //下面这四个回调函数是组件使用者进行设置的回调函数！
    using ConnectedCallback = std::function<void(const PtrConnection&)>;
    using MessageCallback = std::function<void(const PtrConnection&, Buffer *)>; // 业务处理函数！
    using ClosedCallback = std::function<void(const PtrConnection&)>;            // 关闭的时候的处理函数！
    using AnyEventCallback = std::function<void(const PtrConnection&)>;          // 任意事件触发后锁调用的含糊！
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _event_callback;

    //这是是组件内的连接关闭回调！——这是由组件本身设置的！
    //因为服务器组件会将连接管理起来！一旦连接关闭也应当使用组件内的删除方式！
    ClosedCallback  _server_closed_callback;
private:

    //五个channel的回调函数！
    //描述符触发可读事件后调用的函数！
    //作用是接收socket数据放到接收缓冲区中！然后调用Message_callback进行业务处理！
    void HandleRead()
    {
        //1、解锁数据放到缓冲区中！
        char buffer[65536];
        ssize_t ret = _socket.NonBlockRecv(buffer,sizeof(buffer)-1);

        if(ret < 0)
        {
            //出错了不能直接关闭连接还要看一下发送缓冲区有没有数据没有发送！
            //接收缓冲区有没有数据没有处理！
            return ShutdownInLoop();
        }
        else 
        {
            // 没有读取到数据！而不是连接断开！出错是返回-1
            _in_buffer.WriteAndPush(buffer,ret);

        }
        //2.调用Message_callback进行业务处理！
        if(_in_buffer.ReadAbleSize() > 0)
        {
            return _message_callback(shared_from_this(),&_in_buffer);
        }



    }

    // 描述符可写事件触发后调用的函数！
    // 作用是将发送缓冲区的进行发送
    void HandleWrite()
    {
        //如果当前是连接待关闭状态，则有数据就发送完数据再释放连接，没有数据则直接释放！
        //将发送缓冲区的数据进行一次发送！

        ssize_t ret = _socket.NonBlockSend(_out_buffer.GetReadPosition(),_out_buffer.ReadAbleSize());

        if(ret < 0)
        {
            // 发送错误！要关闭连接！但是关闭连接之前要首先将接收缓冲区的数据拿过来！
            if (_in_buffer.ReadAbleSize() > 0)
            {
                _message_callback(shared_from_this(), &_in_buffer);
            }
            return ReleaseInLoop(); // 实际的关闭释放操作！
        }


        _out_buffer.MoveReadOffset(ret); // 要记得将读偏移向后移动！
        if (_out_buffer.ReadAbleSize() == 0)
        {
            _channel.DisabelWrite();
            //如果发送缓冲区为0,那么就关闭写事件监控（用于发送的）
            //避免触发因为在水平触发的情况下，只要缓冲区有剩余空间就会一直触发实践！
            if (_status == DISCONNECTING)
            {
                return ReleaseInLoop();
            }
        }

        return;
    }

    //描述符触发挂断事件后的回调函数！
    void HandleClose()
    {
        //一旦连接挂断了！就什么都处理不了了！如果有数据我们就先要处理一下！
        //处理完毕后再关闭连接！
        if (_in_buffer.ReadAbleSize() > 0)
        {
            _message_callback(shared_from_this(), &_in_buffer);
        }
        return ReleaseInLoop();
    }

    //触发出错事件后的回调函数！
    void HandleError()
    {
        return HandleClose();
    }

    //任意事件触发后所要调用的回调函数！
    void HandleEvent()
    {
        //任意描述符触发任意事件！
        //1.刷新连接的活跃度——延迟定时销毁任务
        //2.调用组件使用者的任意实践回调！
        if(_enable_inactive_release == true)
        {
            _loop->TimerRefresh(_conn_id);//刷新定时任务！
        }
        if(_event_callback)
        {
            _event_callback(shared_from_this());
        }
    }


    //获取连接之后所处状态下要进行的各种设置(启动读事件监控！调用回调函数！)
    void EstablishedInLoop()
    {
        //1.修改连接状态
        //当前的状态一定是上层的半连接状态！
        assert(_status == CONNECTING);
        _status = CONNECTED;//连接进入已完成状态！

        //为什么不再构造函数那边就设置成CONNECTED呢？
        //因为如果在哪里设置读事件监控，可能会被立刻触发！
        //但是时候如果设置了非活跃销毁就会去延迟定时任务销毁的执行！
        //但是这时候_event_cakkback还没有被设置！那么定时任务也就无法触发！
        //那么逻辑上就出现问题了！所以要放在后面！

        //启动读事件监控！
        _channel.EnableRead();

        //调用回调函数！
        if(_connected_callback) _connected_callback(shared_from_this());
    }

    // 实际的释放接口！
    void ReleaseInLoop()
    {
        //1.修改连接状态！
        _status = DISCONNECTED;

        //移除事件监控！
        _channel.Remove();

        //关闭描述符！
        _socket.Close();

        // //如果当前定时器队列中还有定时销毁任务则取消任务！
        // if (_loop->HasTimer(_conn_id))
        // {
        //     _loop->TimeCancel(_conn_id);
        // }

        //调用关闭回调函数！
        if(_closed_callback)
        {
            _closed_callback(shared_from_this());
        }
        //调用用户设置的回调函数进行一些处理！

        if(_server_closed_callback)
           _server_closed_callback(shared_from_this());
            //移除服务器内部的管理信息！

        //避免先调用服务器的close后导致Connection被释放！
        // 失从而导致无法调用_close_callback
        // 所以要先调用_close_callback后调用_server_close_callback
    }

    void SendInLoop(char* data,size_t len)
    {
        //这个接口不是实际的发送接口！
        //只是将数据放入发送缓冲区中！然后启动写事件监控！
        if (_status == DISCONNECTED)
            return;
        
        _out_buffer.WriteAndPush(data,len);


        if(_channel.WriteAble() == false)//如果监控没有打卡就启动写事件监控！
        {
            _channel.EnableWrite();
        }
    }

    //这个号函数不是实际关闭连接操作！需要判断有没有数据待处理待发送！
    void ShutdownInLoop()
    {
        //设置为半关闭状态！
        _status = DISCONNECTING;
        if(_in_buffer.ReadAbleSize() > 0)
        {
            //有可能存在一种情况就是接收缓冲区的数据是不完整的！但是即使如此也是要的！
            if(_message_callback)
                _message_callback(shared_from_this(),&_in_buffer);
        }
        //要么就是写入数据出错时候关闭！要么就是没有发送数据直接关闭！
        if(_out_buffer.ReadAbleSize()>0)
        {
            if (_channel.ReadAble() == false) // 如果不可泄就启动写事件监控！
                _channel.EnableRead();
        }

        if(_out_buffer.ReadAbleSize() == 0)
            ReleaseInLoop();

    }

    //启动非活跃连接释放
    void EnableInActiveReleaseInLoop(int sec)
    {
        //1.将判断标志置为true
        _enable_inactive_release = true;
        //2.添加定时销毁任务——如果任务已经存在就刷新延迟！
        if(_loop->HasTimer(_conn_id) == true) 
        {
            return _loop->TimerRefresh(_conn_id);
        }

        // 如果任务不存在就添加！
        _loop->TimerAdd(_conn_id,sec,std::bind(&Connection::ReleaseInLoop,this));
    }

    void CancelActiveReleaseInLoop()
    {
        _enable_inactive_release = false;

        if (_loop->HasTimer(_conn_id) == true)
            _loop->TimeCancel(_conn_id);
    }


    //升级和切换协议！

    void UpgradeInLoop(const Any& context,
    const ConnectedCallback& conn,
    const MessageCallback& msg,
    const ClosedCallback& closed,
    const AnyEventCallback& event)
    {
        _context = context;
        _connected_callback = conn;
        _message_callback = msg;
        _closed_callback = closed;
        _event_callback = event;
    }

public:
    Connection(EventLoop *loop, uint64_t conn_id, int sockfd)
        :_loop(loop), _conn_id(conn_id), _sockfd(sockfd),
        _enable_inactive_release(false),_channel(_loop,_sockfd),
        _status(CONNECTING),_socket(sockfd)
    {
        _channel.SetCloseCallback(std::bind(&Connection::HandleClose,this));
        _channel.SetEventCallback(std::bind(&Connection::HandleEvent,this));
        _channel.SetReadCallback(std::bind(&Connection::HandleRead,this));
        _channel.SetWriteCallback(std::bind(&Connection::HandleWrite,this));
        _channel.SetErrorCallback(std::bind(&Connection::HandleError,this));
    }
    ~Connection()
    {
        DBG_LOG("RELEASE CONNECTION %p",this);
    }

    // 返回相关的文件描述符
    int FD()
    {
        return _sockfd;
    }

    // 返回_conn_id
    int ID()
    {
        return _conn_id;
    }

    // 判断是否处于CONNECTED状态
    bool Connected()
    {
        return _status == CONNECTED;
    }

    // 设置上下文！——在链接建立完成(CONNECTING)的时候！
    void SetContext(const Any Context)
    {
        _context = Context;
    }

    // 获取当前上下文！返回的是指针！
    Any* GetContext()
    {
        return &_context;
    }

    //设置回调函数！由用户来进行设置！
    void SetConnectedCallback(const ConnectedCallback& cb)
    {
        _connected_callback = cb;
    }
    void SetMessageCallback(const MessageCallback& cb)
    {
        _message_callback = cb;
    }
    void SetClosedCallback(const ClosedCallback& cb)
    {
        _closed_callback = cb;
    }
    void SetAnyEventCallback(const AnyEventCallback& cb)
    {
        _event_callback = cb;
    }
    void SetServerClosedCallback(const ClosedCallback& cb)
    {
        _server_closed_callback = cb; 
    }

    //连接获取之后要进行各种的设置所使用的接口！
    //例如：给channel设置各种回调，启动读监控等

    void Established()
    {
        _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop,this));
    }

    //发送数据将数据放到发送缓冲区中！然后启动写事件监控！
    void Send(char* data,size_t len)
    {
        _loop->RunInLoop(std::bind(&Connection::SendInLoop,this,data,len));
    }

    //提供给组件使用者的关闭接口！将Connection设置为待关闭
    //不是真正的关闭！因为要判断缓冲区里面还有没有数据要进行处理！有就处理完再关闭！
    void Shutdown()
    {
        _loop->RunInLoop(std::bind(&Connection::ShutdownInLoop,this));
    }

    //启动非活跃连接的超时销毁！并定义多长时间算是非活跃！
    void EnableInActiveRelease(int sec)
    {
        _loop->RunInLoop(std::bind(&Connection::EnableInActiveReleaseInLoop,this,sec));
    }

    //关闭非活跃连接的超时销毁！
    void CancelActiveRelease()
    {
        _loop->RunInLoop(std::bind(&Connection::CancelActiveReleaseInLoop,this));
    }


    //切换协议——即重置上下文和阶段性的处理函数！非线程安全的！——必须是在是线程终调用的！
    //否则可能出现协议切换好了！但是还有的事件是使用的旧的协议！
    void Upgrade(const Any& context,
    const ConnectedCallback& conn,const MessageCallback& msg,
    const ClosedCallback& closed,const AnyEventCallback& event)
    {
        _loop->AssertInLoop();
        _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop,this,context,conn,msg,closed,event));
    }

};

class Acceptor
{
private:
    Socket _socket;//创建监听套接字
    EventLoop * _loop;//对监听套接字进行事件监控
    Channel _channel; // 对监听套接字进行事件监控
    using AcceptCallback = std::function<void(int)>;
    AcceptCallback _accept_callback;
private:
    void HandleRead()//监听套接字的读事件回调函数——获取新链接,调用_accept_callback—进行新链接处理
    {
        int newfd = _socket.Accept();
        if(newfd < 0)//获取新连接失败
            return;
        if(_accept_callback) 
            _accept_callback(newfd);
    }

    int CreateServer(int port)
    {
 
        bool ret = _socket.CreateServer(port);
        assert(ret == true);
        return _socket.FD();
    }
public:
/*不能将启动读事件监控放到构造函数中！必须在设置回调函数后！再去启动！*/
/*否则万一启动监控后立刻有事件，处理的时候回调函数还没有设置！那么就会导致新链接得不到处理！又没有释放导致资源泄露！*/
    Acceptor(EventLoop *loop, int port)
        : _loop(loop), _socket(CreateServer(port)), _channel(loop, _socket.FD())
    {

        _channel.SetReadCallback(std::bind(&Acceptor::HandleRead,this));//绑定回调函数
    }
    void SetAcceptCallback(const AcceptCallback& cb)
    {
        _accept_callback = cb;
    }

    void Listen()
    {
        _channel.EnableRead(); // 启动读事件监控
    }
};